package com.css.resoft.p2pmonitor.test;

import java.io.BufferedInputStream;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.InputStream;
import java.io.PrintStream;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.zip.ZipEntry;
import java.util.zip.ZipFile;
import java.util.zip.ZipInputStream;

import javax.xml.parsers.SAXParser;
import javax.xml.parsers.SAXParserFactory;

import org.apache.hadoop.hbase.thrift2.generated.TColumnValue;
import org.apache.hadoop.hbase.thrift2.generated.THBaseService;
import org.apache.hadoop.hbase.thrift2.generated.TIOError;
import org.apache.hadoop.hbase.thrift2.generated.TPut;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.xml.sax.Attributes;
import org.xml.sax.SAXException;
import org.xml.sax.helpers.DefaultHandler;

import com.css.resoft.p2pmonitor.util.RowkeyUtil;


public class PutData extends DefaultHandler{
	public  boolean isQuote = false;
	public  String currentTag = null;
	
	public  static String datadate = null;
	public  static String orgname = null;
	public  static String orgcode = null;
	public  static String reporter = null;
	public  static String checker = null;
	
	public  static int indexOne = 0;
	public  static int indexTwo = 0;
	public  static int indexThree = 0;
	
	public  Map<String,String> record = new HashMap<String,String>();
	public  Map<String,String> headInfo = new HashMap<String,String>();
	
	List<TPut> putOne = new ArrayList<TPut>();
	List<TPut> putTwo = new ArrayList<TPut>();
	List<TPut> putThree = new ArrayList<TPut>();
	
	public final static String slave1host = "39.107.224.196";
	public final static String slave3host = "39.107.112.148";
	public final static int port = 9090;
	
	static TTransport transport = new TSocket(slave3host,port);
	TProtocol protocol = new TBinaryProtocol(transport);
	THBaseService.Iface client = new THBaseService.Client(protocol);
	
	@Override
	public void startDocument() throws SAXException{
		super.startDocument();
		try {
			transport.open();
			indexOne = 0;
			indexTwo = 0;
			indexThree = 0;
		} catch (TTransportException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void endDocument() throws SAXException{
		super.endDocument();
		
		try {
			client.putMultiple(ByteBuffer.wrap("carnon".getBytes()), putOne);
			client.putMultiple(ByteBuffer.wrap("carnonn".getBytes()), putTwo);
			client.putMultiple(ByteBuffer.wrap("carnonnn".getBytes()), putTwo);
		} catch (TIOError e) {
			e.printStackTrace();
		} catch (TException e) {
			e.printStackTrace();
		}
		
		transport.close();
	}
	
	@Override
	public void startElement(String arg0,String arg1,String name,
			Attributes attribtes) throws SAXException{
		super.startElement(arg0, arg1, name, attribtes);
		
		if(name.equals("record")){
			record = new HashMap<String,String>();
		}
		
		isQuote = true;
		currentTag = name;
		
	}
	
    @Override
    public void endElement(String arg0, String arg1, String name) throws SAXException {
        super.endElement(arg0, arg1, name);
        
        if(name.equals("record")){
        	try {
        		TPut put = new TPut();
        		TColumnValue columnValue = null;
        		List<TColumnValue> columnValues = new ArrayList<TColumnValue>();
        		
        		String table = "";
        		
				for(Map.Entry<String, String> entry : record.entrySet()){
					String key = entry.getKey();
					String value = entry.getValue();
					String qualifier = null;
					
					int index = key.indexOf("090");
					if(index > 0){
						qualifier = key.substring(0,index);
						table = key.substring(index, key.length());
					}else{
						qualifier = key;
					}
					
					columnValue = new TColumnValue();
					columnValue.setFamily("info".getBytes());
					columnValue.setQualifier(qualifier.getBytes());
					columnValue.setValue(value.getBytes());
					
					columnValues.add(columnValue);
//	        		System.out.println("key: "+qualifier+" value: "+entry.getValue());
	        	}
				//  do not hit me   i am sorry 
				columnValue = new TColumnValue();
				columnValue.setFamily("info".getBytes());
				columnValue.setQualifier("DATADATE".getBytes());
				columnValue.setValue(datadate.getBytes());
				columnValues.add(columnValue);
//				System.out.println("key = datadate  value = "+datadate);
				
				columnValue = new TColumnValue();
				columnValue.setFamily("info".getBytes());
				columnValue.setQualifier("ORGCODE".getBytes());
				columnValue.setValue(orgcode.getBytes());
				columnValues.add(columnValue);
//				System.out.println("key = datadate  value = "+datadate);
				
				columnValue = new TColumnValue();
				columnValue.setFamily("info".getBytes());
				columnValue.setQualifier("ORGNAME".getBytes());
				columnValue.setValue(orgname.getBytes());
				columnValues.add(columnValue);
//				System.out.println("key = orgname  value = "+orgname);
				
				columnValue = new TColumnValue();
				columnValue.setFamily("info".getBytes());
				columnValue.setQualifier("REPORTER".getBytes());
				columnValue.setValue(reporter.getBytes());
				columnValues.add(columnValue);
//				System.out.println("key = reporter  value = "+reporter);
				
				columnValue = new TColumnValue();
				columnValue.setFamily("info".getBytes());
				columnValue.setQualifier("CHECKER".getBytes());
				columnValue.setValue(checker.getBytes());
				columnValues.add(columnValue);
//				System.out.println("key = checker  value = "+checker);
				
				put.setColumnValues(columnValues);
				byte [] row = null;
				
				if(table.equalsIgnoreCase("0901")){
					indexOne++;
					row = RowkeyUtil.getGetRowkey(orgcode,datadate,indexOne);
					put.setRow(row);
					putOne.add(put);
				}else if(table.equalsIgnoreCase("0902")){
					indexOne++;
					row = RowkeyUtil.getGetRowkey(orgcode,datadate,indexTwo);
					put.setRow(row);
					putTwo.add(put);
				}else if(table.equalsIgnoreCase("0903")){
					indexOne++;
					row = RowkeyUtil.getGetRowkey(orgcode,datadate,indexThree);
					put.setRow(row);
					putThree.add(put);
				}else{
					System.out.println("error");
				}
				
				
				if(putOne.size()>1000){
					client.putMultiple(ByteBuffer.wrap("carnon".getBytes()), putOne);
					putOne.clear();
				}
				
				if(putTwo.size()>1000){
					client.putMultiple(ByteBuffer.wrap("carnonn".getBytes()), putOne);
					putTwo.clear();
				}
				
				if(putThree.size()>1000){
					client.putMultiple(ByteBuffer.wrap("carnonnn".getBytes()), putOne);
					putThree.clear();
				}
				
			} catch (Exception e) {
				e.printStackTrace();
			}
        }
        isQuote = false;
    }
  
	@Override
	public void characters(char[] ch,int start,int length) throws SAXException{
		super.characters(ch, start, length);
		String nodeValue = new String(ch,start,length);
		
		if(isQuote){
			
			if(!currentTag.equals("record"))
				record.put(currentTag, nodeValue);
			
			if(currentTag.equals("datadate")){
				datadate = nodeValue;
			}
			
			// need to improve later  with switch 
			
			if(currentTag.equals("orgcode")){
				orgcode = nodeValue;
			}
			
			if(currentTag.equals("orgname")){
				orgname = nodeValue;
			}
			
			if(currentTag.equals("reporter")){
				reporter = nodeValue;
			}
			
			if(currentTag.equals("checker")){
				checker = nodeValue;
			}
		}
		
	}
	
	public static void getFileList(String fileDir, List<String> filePaths) {
		File file = new File(fileDir);
		if(file.isFile() && fileDir.endsWith("zip")){
			filePaths.add(fileDir);
		}else if(file.isDirectory()){
			for(String str: file.list()){
				getFileList(fileDir+"\\"+str,filePaths);
			}
		}
	}
	
	
	
	public static void main(String[] args) {
		long startTime = System.currentTimeMillis();
		int total = 0;
		String fileDir = "F:\\workspace\\data\\p2p\\";
		List<String> filePaths = new ArrayList<String>();
		int failnum = 0;
		int sucnum = 0;
		try {
			SAXParserFactory factory = SAXParserFactory.newInstance();
			SAXParser parser = factory.newSAXParser();
			PutData handle = new PutData();
			
			getFileList(fileDir,filePaths);
			
			for(String file :filePaths){
				System.out.print(++total + " ");
				try {
					ZipFile zf = new ZipFile(file);
					ZipInputStream zin = new ZipInputStream(new BufferedInputStream(new FileInputStream(file)));
					ZipEntry ze = null;
					while((ze = zin.getNextEntry())!= null){
						
						if(ze.isDirectory()){
							System.out.println(file + " is directory ");
						}else{
							zin.closeEntry();
							
							long size = ze.getSize();
							if(size >0){
								InputStream is = null;
								try {
									is = zf.getInputStream(ze);
									parser.parse(is, handle);
									System.out.println(file +" parse ok ");
									sucnum++;
								} catch (Exception e) {
									transport.close();
									byte[] fileContent = null;
									try {
										is = zf.getInputStream(ze);
										fileContent= new byte[(int)size];
										is.read(fileContent);
										String content = new String(fileContent).replaceFirst("UTF8", "UTF-8").replaceFirst("utf8", "utf-8");
										InputStream   ins =   new   ByteArrayInputStream(content.getBytes("UTF-8"));
										parser.parse(ins, handle);
										System.out.println(file +" parse ok ");
										sucnum ++;
										ins.close();
									} catch (Exception e2) {
										String logFileName = zf.getName();
										String fileName = logFileName.substring(logFileName.lastIndexOf("\\")+1).replace(".zip", "");
										String logDir = "D:\\err\\";
										e2.printStackTrace(new PrintStream(new File(logDir+fileName+".txt")));
										FileOutputStream fos = new FileOutputStream(logDir+fileName);
										fos.write(fileContent);
										fos.flush();
										fos.close();
										
										System.out.println(file+" failed prase");
										failnum++;
										transport.close();
									}
								}finally{
									is.close();
								}
							}else{
								System.out.println(file+" size < 0 ");
							}
						}
					}
					
					
					zin.close();
					zf.close();
				} catch (Exception e) {
					System.out.println(file + " open  error");
					failnum ++;
				}
			}
			
			long endTime = System.currentTimeMillis();
			
			System.out.println("sucnum: "+sucnum);
			System.out.println("failnum: "+failnum);
			
			System.out.println("totalnum: " + filePaths.size());
			
			System.out.println(endTime-startTime);
		} catch (Exception e) {
			e.printStackTrace();
		}
		
	}
	
}
